import lombok.Data;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;


/**
 * 建表语句
 -- source.student definition

 CREATE TABLE `student` (
 `id` varchar(100) COLLATE utf8mb4_bin DEFAULT NULL,
 `name` varchar(100) COLLATE utf8mb4_bin DEFAULT NULL,
 `createtime` varchar(100) COLLATE utf8mb4_bin DEFAULT NULL
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

 INSERT INTO source.student
 (id, name, createtime)
 VALUES('1', '1', '1');

 */
public class MySQLDemo1 {
    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Row> dataInput = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
                // 配置数据库连接信息
                .setDrivername("com.mysql.jdbc.Driver")  // JDBC驱动名
                // com.mysql.jdbc.Driver：是 mysql-connector-java 5 中的
                // com.mysql.cj.jdbc.Driver：是 mysql-connector-java 6及以上 中的
                .setDBUrl("jdbc:mysql://localhost:3306/source?serverTimezone=GMT%2B8&useSSL=false")  // 数据库URL
                // jdbc:mysql://主机名:端口号/数据库名
                // serverTimezone=GMT%2B8：指定时区，设置为北京时间东八区
                // useSSL=false：MySQL 8.0以上版本 不需要建立SSL连接，需要显示关闭
                .setUsername("root")  // 用户名
                .setPassword("root")  // 登录密码
                .setQuery("select id,name,createtime from student")  // 需要执行的SQL语句
                .setRowTypeInfo(new RowTypeInfo(  // 设置查询的列的类型
                        BasicTypeInfo.STRING_TYPE_INFO,  // id：Int类型
                        BasicTypeInfo.STRING_TYPE_INFO,  // name：String类型
                        BasicTypeInfo.STRING_TYPE_INFO))  // age：Int类型
                .finish());
        DataSet<Student> dataMap = dataInput.map(new MapFunction<Row, Student>() {
            @Override
            public Student map(Row row) throws Exception {  // 转换为Student类型
                return new Student(
                        (String) row.getField(0),
                        (String) row.getField(1),
                        (String) row.getField(2));
            }
        });
        dataMap.print();
//        env.execute();

    }
    @Data  // 注解在类上，为类提供读写属性，还提供equals()、hashCode()、toString()方法
    @AllArgsConstructor  // 注解在类上，为类提供全参构造函数，参数的顺序与属性定义的顺序一致
    @NoArgsConstructor  // 注解在类上，为类提供无参构造函数
    public static class Student {
        private String id;
        private String name;
        private String createtime;
    }

}


